package drds.server.route.util;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import drds.server.Server;
import drds.server.cache.LayeredCache;
import drds.server.config.model.Schema;
import drds.server.config.model.ShardRule;
import drds.server.config.model.Table;
import drds.server.execute_engine.data_handling.Value;
import drds.server.route.Route;
import drds.server.route.RouteNode;
import drds.server.route.parser.core.ParserResult;
import drds.server.route.parser.core.TableColumnValueSetMap;
import drds.server.route.sequence.SequenceSql;
import drds.server.route.shard_algorithm.AbstractShardAlgorithm;
import drds.server.server.SessionContext;
import drds.server.server.parser.ServerParse;
import drds.server.sqlparser.parser.StatementParser;
import drds.server.sqlparser.sqlobject.expression.Expression;
import drds.server.sqlparser.sqlobject.statement.insert.InsertStatement;

public class RouteUtil {

	private static final Logger LOGGER = LoggerFactory.getLogger(RouteUtil.class);

	/**
	 * <pre>
	 * ======#======#======#======#======#======#======#======#======#======
	 * 规则计算
	 * ======#======#======#======#======#======#======#======#======#======
	 * </pre>
	 */
	/**
	 * 规则计算。DRDS需要构建list元素list模式的路由字典(具体实现java内存采用treeMap的模式进行处理)
	 */
	public static Set<String> shardRuleCalculate(Table table, Set<Value> valueSet) {

		ShardRule shardRule = table.getShardRule();
		AbstractShardAlgorithm shardAlgorithm = shardRule.getShardAlgorithm();
		String columnName = table.getShardRule().getColumnName();
		Set<String> dataNodeSet = new LinkedHashSet<String>();

		for (Value value : valueSet) {
			if (value.columnValue != null) {
				Integer nodeIndex = shardAlgorithm.calculate(value.columnValue);
				if (nodeIndex == null) {
					throw new IllegalArgumentException("can't find datanode for sharding column:" + columnName + " val:" + value.columnValue);
				} else {
					String dataNode = table.getDataNodeList().get(nodeIndex);
					dataNodeSet.add(dataNode);
					// 只能是一个结点
					value.setNodeIndex(nodeIndex);
				}
			} else if (value.rangeValue != null) {
				Integer[] nodeIndexs = shardAlgorithm.calculateRange(String.valueOf(value.rangeValue.beginValue), String.valueOf(value.rangeValue.endValue));
				if (nodeIndexs != null) {
					/**
					 * 不能确认 colPair的 nodeid是否会有其它影响
					 */
					if (nodeIndexs.length == 0) {
						throw new RuntimeException("没有找到对应的节点,此处是业务错误需要报告给客户端和监控平台");
					} else {
						ArrayList<String> dataNodeList = table.getDataNodeList();
						String dataNode = null;
						for (Integer nodeIndex : nodeIndexs) {
							dataNode = dataNodeList.get(nodeIndex);
							dataNodeSet.add(dataNode);
						}
					}
				}
			}

		}
		return dataNodeSet;
	}

	/**
	 * <pre>
	 * ======#======#======#======#======#======#======#======#======#======
	 * 路由
	 * ======#======#======#======#======#======#======#======#======#======
	 * </pre>
	 */
	/**
	 * 随机路由节点
	 */
	public static void routeToRandomNode(//
			Route route,//
			Schema schema,//
			String tableName,//
			String sql) {//

		String randomNodeIndex = getRandomNodeIndex(schema, tableName);
		/** RouteNode */
		RouteNode routeNode = new RouteNode(randomNodeIndex, route.getSqlType(), sql);
		if (route.getWriteTrue$ReadFalse() != null) {
			routeNode.setWriteTrue$ReadFalse(route.getWriteTrue$ReadFalse());
		}
		if (route.getMasterTrue$SlaveFalse() != null) {
			routeNode.setMasterTrue$SlaveFalse(route.getMasterTrue$SlaveFalse());
		}
		routeNode.setRoute(route);
		/** routeNodes */
		RouteNode[] routeNodes = new RouteNode[1];
		routeNodes[0] = routeNode;
		route.setRouteNodes(routeNodes);
	}

	/**
	 * 根据表名随机获取一个节点
	 */
	private static String getRandomNodeIndex(Schema schema, String tableName) {
		Map<String, Table> tableMap = schema.getTableMap();
		Table table = tableMap.get(tableName);
		if (table == null) {
			throw new NullPointerException("table");
		}
		return table.getRandomNodeIndex();
	}

	/**
	 * 路由到一个节点
	 */
	public static Route routeToSingleNode(//
			Route route,//
			String dataNodeId,//
			String sql) {//
		if (route == null) {
			throw new NullPointerException("routeContext");
		}
		if (dataNodeId == null) {
			throw new NullPointerException("dataNodeId");
		}
		if (sql == null) {
			throw new NullPointerException("sql");
		}
		//
		RouteNode routeNode = new RouteNode(dataNodeId, route.getSqlType(), sql);
		routeNode.setRoute(route);
		if (route.getWriteTrue$ReadFalse() != null) {
			routeNode.setWriteTrue$ReadFalse(route.getWriteTrue$ReadFalse());
		}
		if (route.getMasterTrue$SlaveFalse() != null) {
			routeNode.setMasterTrue$SlaveFalse(route.getMasterTrue$SlaveFalse());
		}
		//
		RouteNode[] routeNodes = new RouteNode[1];
		routeNodes[0] = routeNode;
		route.setRouteNodes(routeNodes);
		route.setFinishedRoute(true);

		return route;
	}

	/**
	 * 路由到多个节点
	 */
	public static Route routeToMultiNode(Route route, Collection<String> dataNodes, String sql, boolean isGlobalTable, boolean cache) {

		route = RouteUtil.routeToMultiNode(route, dataNodes, sql, cache);
		route.setGlobalTable(isGlobalTable);
		return route;
	}

	public static Route routeToMultiNode(Route route, Collection<String> dataNodes, String sql, boolean cache) {
		RouteNode[] routeNodes = new RouteNode[dataNodes.size()];
		int i = 0;
		for (String dataNode : dataNodes) {
			RouteNode routeNode = new RouteNode(dataNode, route.getSqlType(), sql);
			if (route.getWriteTrue$ReadFalse() != null) {
				routeNode.setWriteTrue$ReadFalse(route.getWriteTrue$ReadFalse());
			}
			if (route.getMasterTrue$SlaveFalse() != null) {
				routeNode.setMasterTrue$SlaveFalse(route.getMasterTrue$SlaveFalse());
			}
			routeNode.setRoute(route);
			routeNodes[i++] = routeNode;
		}
		route.setCache(cache);
		route.setRouteNodes(routeNodes);
		return route;
	}

	/**
	 * <pre>
	 * ======#======#======#======#======#======#======#======#======#======
	 * 插入操作
	 * ======#======#======#======#======#======#======#======#======#======
	 * </pre>
	 */

	public static boolean processInsert(SessionContext sessionContext, Schema schema, String sql, int sqlType) throws IllegalArgumentException {
		String tableName = getTableNameOfInsertSql(sql);
		Table table = schema.getTableMap().get(tableName);
		boolean processedInsert = false;
		// 判断是有自增字段
		if (null != table && table.isUseGlobalSequence()) {
			String primaryKey = table.getPrimaryKeyName();
			processedInsert = processInsert(sessionContext, schema, sql, sqlType, tableName, primaryKey);
		}
		return processedInsert;
	}

	public static boolean processInsert(//
			SessionContext sessionContext,//
			Schema schema, //
			String sql, //
			int sqlType, //
			String tableName,//
			String primaryKeyName) throws IllegalArgumentException {

		int firstLeftBracketIndex = sql.indexOf("(");
		int firstRightBracketIndex = sql.indexOf(")");
		int valuesIndex = sql.indexOf("values");
		//
		int selectIndex = sql.indexOf("select");
		int fromIndex = sql.indexOf("from");
		// 屏蔽insert into table1 select * from table2语句
		if (firstLeftBracketIndex < 0) {
			String msg = "invalid sql:" + sql;
			LOGGER.error(msg);
			throw new IllegalArgumentException(msg);
		}
		// 屏蔽批量插入
		if (selectIndex > 0 && //
				fromIndex > 0 && //
				selectIndex > firstRightBracketIndex && //
				valuesIndex < 0) {//
			String msg = "multi insert not provided";
			LOGGER.error(msg);
			throw new IllegalArgumentException(msg);
		}
		/**
		 * 插入语句必须提供列结构，因为DRDS默认对于表结构无感知
		 */
		if (valuesIndex + "values".length() <= firstLeftBracketIndex) {
			throw new IllegalArgumentException("insert must provide ColumnList");
		}

		/**
		 * 如果主键不在插入语句的fields中，则需要进一步处理
		 */
		boolean processedInsert = !isPrimaryKeyInColumns(sql, primaryKeyName, firstLeftBracketIndex, firstRightBracketIndex);
		if (processedInsert) {
			List<String> insertSqlList = handleBatchInsert(sql, valuesIndex);
			for (String insertSql : insertSqlList) {
				processInsert(sessionContext, schema, insertSql, sqlType, tableName, primaryKeyName, firstLeftBracketIndex + 1, insertSql.indexOf('(', firstRightBracketIndex) + 1);
			}
		}
		return processedInsert;
	}

	private static boolean isPrimaryKeyInColumns(String sql, String primaryKeyName, int firstLeftBracketIndex, int firstRightBracketIndex) {

		if (primaryKeyName == null) {
			throw new RuntimeException("please make sure the primaryKey's config is not null in schemal.xml");
		}

		boolean isPrimaryKeyInColumns = false;
		String $sql = sql.substring(firstLeftBracketIndex, firstRightBracketIndex + 1);

		int primaryKeyNameLength = primaryKeyName.length();
		int primaryKeyNameIndex = 0;
		int index = 0;
		for (;;) {
			index = $sql.indexOf(primaryKeyName, primaryKeyNameIndex);
			if (index >= 0 && index < firstRightBracketIndex) {
				char $char = $sql.charAt(index - 1);
				if ($char <= ' ' || //
						$char == '`' || //
						$char == ',' || //
						$char == '(') {//
					$char = $sql.charAt(index + primaryKeyName.length());
					isPrimaryKeyInColumns = $char <= ' ' || //
							$char == '`' || //
							$char == ',' || //
							$char == ')';//
				}
				if (isPrimaryKeyInColumns) {
					break;
				}
				primaryKeyNameIndex = index + primaryKeyNameLength;
			} else {
				break;
			}
		}
		return isPrimaryKeyInColumns;
	}

	public static List<String> handleBatchInsert(//
			String sql,//
			int valuesIndex) {//
		List<String> sqlList = new LinkedList<String>();
		String prefix = sql.substring(0, valuesIndex + "values".length());
		String values = sql.substring(valuesIndex + "values".length());
		int flag = 0;
		StringBuilder sb = new StringBuilder();
		sb.append(prefix);
		int length = values.length();
		for (int i = 0; i < length; i++) {
			char $char = values.charAt(i);
			if ($char == '(' && flag == 0) {
				flag = 1;
				sb.append($char);
			} else if ($char == '\"' && flag == 1) {
				flag = 2;
				sb.append($char);
			} else if ($char == '\'' && flag == 1) {
				flag = 2;
				sb.append($char);
			} else if ($char == '\\' && flag == 2) {
				flag = 3;
				sb.append($char);
			} else if (flag == 3) {
				flag = 2;
				sb.append($char);
			} else if ($char == '\"' && flag == 2) {
				flag = 1;
				sb.append($char);
			} else if ($char == '\'' && flag == 2) {
				flag = 1;
				sb.append($char);
			} else if ($char == ')' && flag == 1) {
				flag = 0;
				sb.append($char);
				sqlList.add(sb.toString());
				sb = new StringBuilder();
				sb.append(prefix);
			} else if ($char == ',' && flag == 0) {
				continue;
			} else {
				sb.append($char);
			}
		}
		return sqlList;
	}

	static String sequencePrefix = "next value for sequence_";
	static int sequencePrefixLength = sequencePrefix.length();

	private static void processInsert(//
			SessionContext sessionContext,//
			Schema schema,//
			String sql,//
			int sqlType,//
			String tableName,//
			String primaryKeyName, //
			int afterFirstLeftBracketIndex,//
			int afterLastLeftBracketIndex) {//
		/**
		 * <pre>
		 * 对于主键不在插入语句的fields中的SQL，需要改写。 
		 * 比如hotnews主键为id，插入语句为： insert into hotnews(title) values('aaa'); 
		 * 需要改写成： insert into hotnews(id, title) values(next value for SEQ_hotnews,'aaa');
		 * </pre>
		 */
		int tableNameLength = tableName.length();
		int primaryKeyNameLength = primaryKeyName.length();
		char[] chars = new char[sql.length() + primaryKeyNameLength + sequencePrefixLength + tableNameLength + 2];
		// insert into table(
		sql.getChars(0, afterFirstLeftBracketIndex, chars, 0);
		// primaryKeyName
		int index = afterFirstLeftBracketIndex;
		primaryKeyName.getChars(0, primaryKeyNameLength, chars, index);
		index += primaryKeyNameLength;
		// ,
		chars[index] = ',';
		index++;
		// afterLastLeftBracketIndex - afterFirstLeftBracketIndex;
		sql.getChars(afterFirstLeftBracketIndex, afterLastLeftBracketIndex, chars, index);
		index += afterLastLeftBracketIndex - afterFirstLeftBracketIndex;
		//
		sequencePrefix.getChars(0, sequencePrefixLength, chars, index);
		index += sequencePrefixLength;
		tableName.getChars(0, tableNameLength, chars, index);
		index += tableNameLength;
		// ,
		chars[index] = ',';
		index++;
		// afterLastLeftBracketIndex
		sql.getChars(afterLastLeftBracketIndex, sql.length(), chars, index);
		processSequenceSql(sessionContext, schema, new String(chars), sqlType);
	}

	public static boolean processSequence(SessionContext sessionContext, Schema schema, String sql, int sqlType) {
		if (sql.indexOf(" sequence_") != -1) {
			processSequenceSql(sessionContext, schema, sql, sqlType);
			return true;
		} else {
			return false;
		}

	}

	public static void processSequenceSql(SessionContext sessionContext, Schema schema, String sql, int sqlType) {
		SequenceSql sequenceSql = new SequenceSql(sessionContext.getSession(), schema, sql, sqlType);
		Server.getInstance().getSequnceProcessor().addSequenceSql(sequenceSql);
	}

	/**
	 * <pre>
	 * ======#======#======#======#======#======#======#======#======#======
	 * ER操作
	 * ======#======#======#======#======#======#======#======#======#======
	 * </pre>
	 */

	public static Route routeByERRootKey(//
			Route route, //
			Schema schema, //
			Table table,//
			String sql,//
			int sqlType, //
			String rootKeyValue//
	) {

		/**
		 * 使用父表去查询dataNode
		 */
		Set<Value> valueSet = new HashSet<Value>(1);
		Value value = new Value(rootKeyValue);
		valueSet.add(value);
		Set<String> dataNodeSet = shardRuleCalculate(table.getRootParentTable(), valueSet);
		if (dataNodeSet.isEmpty() || dataNodeSet.size() > 1) {
			throw new IllegalArgumentException("parent key can't find  valid datanode ,expect 1 but found: " + dataNodeSet.size());
		}
		String dataNode = dataNodeSet.iterator().next();
		return RouteUtil.routeToSingleNode(route, dataNode, sql);

	}

	public static Set<String> routeByERJoinKeyValue(//
			Route route,//
			Table table,//
			Set<Value> valueSet) {//
		if (valueSet == null || valueSet.size() == 0) {
			throw new NullPointerException("valueSet");
		}
		if (valueSet.size() != 1) {
			String msg = "joinKey can't have multi Value";
			LOGGER.error(msg);
			throw new IllegalArgumentException(msg);
		}
		Set<String> dataNodeSet = new LinkedHashSet<String>();
		if (table.isSecondLevelChildTable() && //
				table.getParentTable().getShardColumnName().equals(table.getParentTableJoinedKeyColumnName())) {
			// using parent rule to find datanode
			Set<String> $dataNodeSet = shardRuleCalculate(table.getParentTable(), valueSet);
			if ($dataNodeSet.isEmpty()) {
				throw new IllegalArgumentException("parent key can't find  valid datanode ,expect 1 but found: " + $dataNodeSet.size());
			}
			dataNodeSet.addAll($dataNodeSet);
			return dataNodeSet;
		} else {
			// 该处存在疑问
			dataNodeSet.addAll(table.getParentTable().getDataNodeList());
		}

		return dataNodeSet;
	}

	public static boolean processERChildTable(//
			final SessionContext sessionContext,//
			final Schema schema,//
			final String sql) {
		String tableName = getTableNameOfInsertSql(sql);
		final Table table = schema.getTableMap().get(tableName);
		// 判断是否为子表，如果不是，只会返回false
		if (null != table && !table.isChildTable()) {
			return false;
		} else {
			final Route route = new Route(sql, ServerParse.INSERT);
			String rootKeyColumnName = table.getRootParentTableKeyColumnName();//
			InsertStatement insertStatement = (InsertStatement) (new StatementParser(sql)).parseInsertStatement();
			// 判断条件完整性，取得解析后语句列中的joinkey列的index
			int rootKeyIndex = getJoinKeyIndex(insertStatement.getColumnNameList(), rootKeyColumnName);
			if (rootKeyIndex == -1) {
				String inf = "rootKey not provided :" + table.getJoinKeyColumnName() + "," + insertStatement;
				LOGGER.error(inf);
				throw new IllegalArgumentException(inf);
			}
			// 子表不支持批量插入
			String rootKeyValue = insertStatement.getValueList().//
					getValueList().get(rootKeyIndex).toString();

			String $sql = insertStatement.toString();

			// try to route by ER parent partion key
			// 如果是二级子表（父表不再有父表）,并且分片字段正好是joinkey字段，调用routeByERParentKey
			Route $route = RouteUtil.routeByERRootKey(//
					route,//
					schema,//
					table,//
					$sql, //
					ServerParse.INSERT,//
					rootKeyValue);//
			if ($route != null) {
				boolean processedInsert = false;
				// 判断是否需要全局序列号
				if (sessionContext != null && table.isUseGlobalSequence()) {
					String primaryKeyName = table.getPrimaryKeyName();
					processedInsert = processInsert(//
							sessionContext,//
							schema, //
							$sql,//
							ServerParse.INSERT, //
							table.getName(),//
							primaryKeyName);//
				}
				if (processedInsert == false) {
					route.setFinishedRoute(true);
					sessionContext.getSession().execute(route, ServerParse.INSERT);
				}
				return true;
			} else {
				throw new IllegalArgumentException("没有找到切分数据库和相关的表的路由信息");
			}

		}
	}

	/**
	 * 寻找joinKey的索引
	 * 
	 * @return -1表示没找到
	 */
	private static int getJoinKeyIndex(List<Expression> columnNameList, String joinKeyName) {
		int size = columnNameList.size();
		for (int i = 0; i < size; i++) {
			String columnName = columnNameList.get(i).toString();
			if (columnName.equals(joinKeyName)) {
				return i;
			}
		}
		return -1;
	}

	/**
	 * <pre>
	 * ======#======#======#======#======#======#======#======#======#======
	 * 表路由
	 * ======#======#======#======#======#======#======#======#======#======
	 * </pre>
	 */
	/**
	 * 单表路由
	 */
	protected static Route singleTableRoute(//
			Route route,//
			Schema schema,//
			ParserResult parserResult,//
			String tableName,//
			TableColumnValueSetMap tableColumnValueSetMap,//
			boolean isSelect,//
			LayeredCache cache) {//

		Table table = schema.getTableMap().get(tableName);
		if (table == null) {
			throw new NullPointerException("table:" + tableName);
		}
		/** 全局表 */
		if (table.isGlobalTable()) {
			if (isSelect) {
				// global select ,not cache route result
				route.setCache(false);
				return RouteUtil.routeToSingleNode(route, table.getRandomNodeIndex(), parserResult.getSql());
			} else {
				// insert into 全局表的记录
				/**
				 * 默认指定到DRDS那张可以扩散和对比的那张表的节点，这个和不一样
				 */
				// 暂时代码是路由到多数据库(一个数据库只有一个全局表,他依赖于其相关的一张表)
				return RouteUtil.routeToMultiNode(route, table.getDataNodeList(), parserResult.getSql(), true, false);
			}
		} else {// 单表或者分库表
			if (!findShardkey(schema, table, tableColumnValueSetMap)) {
				throw new IllegalArgumentException("route rule for table " + table.getName() + " is required: " + parserResult.getSql());
			}
			if (table.getShardColumnName() == null && //
					!table.isSecondLevelChildTable()) {// 单表且不是childTable
				return RouteUtil.routeToMultiNode(route, table.getDataNodeList(), parserResult.getSql(), route.isCache());
			} else {
				// 每个表对应的路由映射
				Map<String, Set<String>> tableDataNodeSetMap = new HashMap<String, Set<String>>();
				if (tableColumnValueSetMap.getTableColumnValueSetMapMap() != null && tableColumnValueSetMap.getTableColumnValueSetMapMap().size() > 0) {
					RouteUtil.routeTables(route, schema, parserResult.getSql(), isSelect, tableColumnValueSetMap.getTableColumnValueSetMapMap(), tableDataNodeSetMap, cache);
					if (route.isRouteOk()) {
						return route;
					}
				}

				if (tableDataNodeSetMap.get(tableName) == null) {
					return RouteUtil.routeToMultiNode(route, table.getDataNodeList(), parserResult.getSql(), route.isCache());
				} else {
					return RouteUtil.routeToMultiNode(route, tableDataNodeSetMap.get(tableName), parserResult.getSql(), route.isCache());
				}
			}
		}
	}

	/**
	 * 多表路由
	 */
	public static Route multiTableRoute(//
			Route route,//
			ParserResult parseResult,//
			Schema schema,//
			TableColumnValueSetMap tableColumnValueSetMap,//
			boolean isSelect,//
			LayeredCache cache//
	) {//

		List<String> tableNameList = parseResult.getTableNameList();
		// 只有一个表的
		if (tableNameList.size() == 1) {
			return RouteUtil.singleTableRoute(route, schema, parseResult, tableNameList.get(0), tableColumnValueSetMap, isSelect, cache);
		}
		// 每个表对应的路由映射
		Map<String, Set<String>> tableNodeSetMap = new HashMap<String, Set<String>>();
		// 分库解析信息不为空
		Map<String, Map<String, Set<Value>>> tableColumnValueSetMapMap = tableColumnValueSetMap.getTableColumnValueSetMapMap();
		if (tableColumnValueSetMapMap != null && tableColumnValueSetMapMap.size() > 0) {
			// 为分库表找路由
			RouteUtil.routeTables(route, schema, parseResult.getSql(), isSelect, tableColumnValueSetMapMap, tableNodeSetMap, cache);
			if (route.isRouteOk()) {
				return route;
			}
		}

		// 为全局表和单库表找路由
		for (String tableName : tableNameList) {
			Table table = schema.getTableMap().get(tableName);
			if (table == null) {
				String msg = "can't find table define in schema " + tableName + " schema:" + schema.getName();
				LOGGER.error(msg);
				throw new IllegalArgumentException(msg);
			}
			if (table.isGlobalTable()) {
				// 全局表-如果在前面没有数据节点则添加该表的所有的数据节点
				if (tableNodeSetMap.get(tableName) == null) {
					tableNodeSetMap.put(tableName, new HashSet<String>());
				}
				tableNodeSetMap.get(tableName).addAll(table.getDataNodeList());
			} else if (tableNodeSetMap.get(tableName) == null) {
				// 余下的表都是单库表-如果在前面没有数据节点则添加该表的所有的数据节点
				tableNodeSetMap.put(tableName, new HashSet<String>());
				tableNodeSetMap.get(tableName).addAll(table.getDataNodeList());
			}
		}
		//
		Set<String> dataNodeSet = new HashSet<String>();
		boolean isFirstAdd = true;
		for (Map.Entry<String, Set<String>> entry : tableNodeSetMap.entrySet()) {
			if (entry.getValue() == null || entry.getValue().size() == 0) {
				throw new IllegalArgumentException("parent key can't find any valid datanode ");
			} else {
				if (isFirstAdd) {
					dataNodeSet.addAll(entry.getValue());
					isFirstAdd = false;
				} else {
					dataNodeSet.retainAll(entry.getValue());
					// 两个表的路由无交集
					if (dataNodeSet.size() == 0) {
						String errMsg = "invalid route in sql, multi tables found but datanode has no intersection " + " sql:" + parseResult.getSql();
						LOGGER.error(errMsg);
						throw new IllegalArgumentException(errMsg);
					}
				}
			}
		}// 求交集

		if (dataNodeSet != null && dataNodeSet.size() > 0) {
			if (dataNodeSet.size() > 1 && allIsGlobalTable(parseResult, schema)) {
				// mulit routes ,not cache route result
				if (isSelect) {
					route.setCache(false);
					RouteUtil.routeToSingleNode(route, dataNodeSet.iterator().next(), parseResult.getSql());
				} else {// delete 删除全局表的记录
					RouteUtil.routeToMultiNode(route, dataNodeSet, parseResult.getSql(), true, isSelect);
				}

			} else {
				RouteUtil.routeToMultiNode(route, dataNodeSet, parseResult.getSql(), isSelect);
			}

		}
		return route;

	}

	/**
	 * 处理分库分表路由
	 */
	protected static void routeTables(//
			Route route,//
			Schema schema,//
			String sql,//
			boolean isSelect,//
			Map<String, Map<String, Set<Value>>> tableColumnValueSetMapMap,//
			Map<String, Set<String>> tableNodeSetMap,//
			LayeredCache cache) throws IllegalArgumentException {

		// 为分库表找路由
		for (Map.Entry<String, Map<String, Set<Value>>> entry : tableColumnValueSetMapMap.entrySet()) {
			String tableName = entry.getKey();
			Table table = schema.getTableMap().get(tableName);
			if (table == null) {
				String msg = "can't find table define in schema " + tableName + " schema:" + schema.getName();
				LOGGER.error(msg);
				throw new IllegalArgumentException(msg);
			}
			// 全局表或者不分库的表略过（全局表后面再计算）
			if (table.isGlobalTable() || table.isNotShardTable()) {
				// schema.getTableMap().get(tableName).getDataNodeList().size()
				// == 1
				// 系统所有的表都需要进行分库分表。或者指定不进行分库分表的表示。

				// 1需要判断当前节点是否含有全局表。
				// 2未进行切分的表不能和切分表进行一起计算。
				continue;
			} else {// 非全局表：分库表、childTable、其他
				Map<String, Set<Value>> columnValueSetMap = entry.getValue();
				String primaryKeyName = table.getPrimaryKeyName();
				String joinKeyColumnName = table.getJoinKeyColumnName();
				//
				String shardColumnName = table.getShardColumnName();
				boolean isFoundShardValue = shardColumnName != null && entry.getValue().get(shardColumnName) != null;
				if (entry.getValue().get(primaryKeyName) != null) {
					// 主键查找
					// try by primary key if found in cache
					Set<Value> primaryKeyValueSet = entry.getValue().get(primaryKeyName);
					//
					if (primaryKeyValueSet != null && primaryKeyValueSet.size() > 0) {
						String tableNameKey = tableName;
						boolean allFound = true;
						for (Value value : primaryKeyValueSet) {// 可能id
							// in(1,2,3)多主键
							String columnValueKey = value.columnValue;
							String dataNode = (String) cache.get(tableNameKey, columnValueKey);
							if (dataNode == null) {
								allFound = false;
								continue;
							} else {
								if (tableNodeSetMap.get(tableName) == null) {
									tableNodeSetMap.put(tableName, new HashSet<String>());
								}
								tableNodeSetMap.get(tableName).add(dataNode);
								continue;
							}
						}
						if (!allFound) {
							// need cache primary key ->datanode relation
							if (isSelect && table.getPrimaryKeyName() != null) {
								route.setPrimaryKey(tableNameKey + '.' + table.getPrimaryKeyName());
							}
						} else {// 主键缓存中找到了就执行循环的下一轮
							continue;
						}
					}
				}
				if (isFoundShardValue) {// 分库表
					Set<Value> shardColumnValueSet = columnValueSetMap.get(shardColumnName);
					if (shardColumnValueSet == null || //
							shardColumnValueSet.size() == 0) {
						if (tableNodeSetMap.get(tableName) == null) {
							tableNodeSetMap.put(tableName, new HashSet<String>());
						}
						tableNodeSetMap.get(tableName).addAll(table.getDataNodeList());
					} else {
						for (Value value : shardColumnValueSet) {
							if (value.columnValue != null) {
								Integer nodeIndex = table.getShardRule().getShardAlgorithm().calculate(value.columnValue);
								if (nodeIndex == null) {
									String msg = "can't find any valid datanode :" + table.getName() + " -> " + table.getShardColumnName() + " -> " + value.columnValue;
									LOGGER.error(msg);
									throw new IllegalArgumentException(msg);
								}

								ArrayList<String> dataNodeList = table.getDataNodeList();
								String dataNode;
								if (nodeIndex >= 0 && nodeIndex < dataNodeList.size()) {
									dataNode = dataNodeList.get(nodeIndex);
								} else {
									dataNode = null;
									String msg = "Can't find a valid data node for specified node index :" + table.getName() + " -> " + table.getShardColumnName() + " -> " + value.columnValue + " -> " + "Index : " + nodeIndex;
									LOGGER.error(msg);
									throw new IllegalArgumentException(msg);
								}
								if (dataNode != null) {
									if (tableNodeSetMap.get(tableName) == null) {
										tableNodeSetMap.put(tableName, new HashSet<String>());
									}
									tableNodeSetMap.get(tableName).add(dataNode);
								}
							}
							if (value.rangeValue != null) {
								Integer[] nodeIndexs = table.getShardRule().getShardAlgorithm().calculateRange(value.rangeValue.beginValue.toString(), value.rangeValue.endValue.toString());
								ArrayList<String> dataNodeList = table.getDataNodeList();
								String dataNode;
								for (Integer nodeIndex : nodeIndexs) {
									if (nodeIndex >= 0 && nodeIndex < dataNodeList.size()) {
										dataNode = dataNodeList.get(nodeIndex);
									} else {
										String msg = "Can't find valid data node(s) for some of specified node indexes :" + table.getName() + " -> " + table.getShardColumnName();
										LOGGER.error(msg);
										throw new IllegalArgumentException(msg);
									}
									if (dataNode != null) {
										if (tableNodeSetMap.get(tableName) == null) {
											tableNodeSetMap.put(tableName, new HashSet<String>());
										}
										tableNodeSetMap.get(tableName).add(dataNode);

									}
								}
							}
						}
					}
				} else if (joinKeyColumnName != null && columnValueSetMap.get(joinKeyColumnName) != null && columnValueSetMap.get(joinKeyColumnName).size() != 0) {
					// childTable
					// (如果是select
					// 语句的父子表join)之前要找到root
					// table,将childTable移除,只留下root
					// table
					Set<Value> joinKeyColumnValueSet = columnValueSetMap.get(joinKeyColumnName);

					Set<String> dataNodeSet = routeByERJoinKeyValue(route, table, joinKeyColumnValueSet);

					if (dataNodeSet.isEmpty()) {
						throw new IllegalArgumentException("parent key can't find any valid datanode ");
					}
					if (dataNodeSet.size() > 1) {
						RouteUtil.routeToMultiNode(route, dataNodeSet, sql, route.isCache());
						route.setFinishedRoute(true);
						return;
					} else {
						route.setCache(true);
						RouteUtil.routeToSingleNode(route, dataNodeSet.iterator().next(), sql);
						return;
					}

				} else {
					// 没找到拆分字段，该表的所有节点都路由
					if (tableNodeSetMap.get(tableName) == null) {
						tableNodeSetMap.put(tableName, new HashSet<String>());
					}
					tableNodeSetMap.get(tableName).addAll(table.getDataNodeList());
				}
			}
		}
	}

	public static boolean allIsGlobalTable(ParserResult parserResult, Schema schema) {
		for (String tableName : parserResult.getTableNameList()) {
			Table table = schema.getTableMap().get(tableName);
			if (!(table != null && table.isGlobalTable())) {
				return false;
			}
		}
		return true;
	}

	/**
	 * 判断切分键是否存在
	 */
	public static boolean findShardkey(Schema schema, Table table, TableColumnValueSetMap tableColumnValueSetMap) {
		String tableName = table.getName();
		if (tableColumnValueSetMap.getTableColumnValueSetMapMap().get(tableName) == null || //
				tableColumnValueSetMap.getTableColumnValueSetMapMap().get(tableName).size() == 0) {//
			return false;
		} else {
			boolean hasRequiredValue = false;
			for (Map.Entry<String, Set<Value>> entry : tableColumnValueSetMap.getTableColumnValueSetMapMap().get(tableName).entrySet()) {
				String columnName = entry.getKey();
				// 条件字段是拆分字段
				if (columnName.equals(table.getShardColumnName())) {
					hasRequiredValue = true;
					break;
				}
			}
			return hasRequiredValue;
		}

	}

	/**
	 * insert into tableName()
	 */
	public static String getTableNameOfInsertSql(String insertSql) {
		String sql = insertSql;
		int pos = 0;
		boolean insertFound = false;
		boolean intoFound = false;
		int tableStartIndx = -1;
		int tableEndIndex = -1;
		while (pos < sql.length()) {
			char ch = sql.charAt(pos);
			// 忽略处理注释 /* */ BEN
			if (ch == '/' && pos + 4 < sql.length() && sql.charAt(pos + 1) == '*') {
				if (sql.substring(pos + 2).indexOf("*/") != -1) {
					pos += sql.substring(pos + 2).indexOf("*/") + 4;
					continue;
				} else {
					// 不应该发生这类情况。
					throw new IllegalArgumentException("sql 注释 语法错误");
				}
			} else if (ch <= ' ' || ch == '(' || ch == '`') {//
				if (tableStartIndx > 0) {
					tableEndIndex = pos;
					break;
				} else {
					pos++;
					continue;
				}
			} else if (ch == 'i' || ch == 'I') {
				if (intoFound) {
					if (tableStartIndx == -1 && ch != '`') {
						tableStartIndx = pos;
					}
					pos++;
				} else if (insertFound) {// into start
					// 必须全部都为INTO才认为是into
					if (pos + 5 < sql.length() && (sql.charAt(pos + 1) == 'n' || sql.charAt(pos + 1) == 'N') && (sql.charAt(pos + 2) == 't' || sql.charAt(pos + 2) == 'T') && (sql.charAt(pos + 3) == 'o' || sql.charAt(pos + 3) == 'O') && (sql.charAt(pos + 4) <= ' ')) {
						pos = pos + 5;
						intoFound = true;
					} else {
						pos++;
					}
				} else {
					// 矫正必须全部都为 INSERT才认为是insert
					// insert start
					if (pos + 7 < sql.length() && (sql.charAt(pos + 1) == 'n' || sql.charAt(pos + 1) == 'N') && (sql.charAt(pos + 2) == 's' || sql.charAt(pos + 2) == 'S') && (sql.charAt(pos + 3) == 'e' || sql.charAt(pos + 3) == 'E') && (sql.charAt(pos + 4) == 'r' || sql.charAt(pos + 4) == 'R') && (sql.charAt(pos + 5) == 't' || sql.charAt(pos + 5) == 'T') && (sql.charAt(pos + 6) <= ' ')) {
						pos = pos + 7;
						insertFound = true;
					} else {
						pos++;
					}
				}
			} else {
				if (tableStartIndx == -1) {
					tableStartIndx = pos;
				}
				pos++;
			}

		}
		return sql.substring(tableStartIndx, tableEndIndex);
	}

}
